package cn.zzdt4j.config.service.biz.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.zzdt4j.common.extension.enums.DelEnum;
import cn.zzdt4j.common.toolkit.*;
import cn.zzdt4j.config.event.LocalDataChangeEvent;
import cn.zzdt4j.config.mapper.ConfigInfoMapper;
import cn.zzdt4j.config.mapper.ConfigInstanceMapper;
import cn.zzdt4j.config.model.*;
import cn.zzdt4j.config.service.ConfigCacheService;
import cn.zzdt4j.config.service.ConfigChangePublisher;
import cn.zzdt4j.config.service.biz.ConfigService;
import cn.zzdt4j.config.service.biz.OperationLogService;
import cn.zzdt4j.core.config.ApplicationContextHolder;
import cn.zzdt4j.server.common.base.exception.ServiceException;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.toolkit.SqlHelper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.util.*;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static cn.zzdt4j.common.constant.MagicNumberConstants.*;
import static cn.zzdt4j.common.executor.support.BlockingQueueTypeEnum.*;
import static cn.zzdt4j.config.service.ConfigCacheService.getContent;

/**
 * The type Config service impl.
 *
 * @author by <a href="mailto:ligang941012@gmail.com">gang.Li</a>
 * @since 2023 /11/6 20:38
 */
@Slf4j
@Service
@AllArgsConstructor
public class ConfigServiceImpl implements ConfigService {

    private final ConfigInfoMapper configInfoMapper;

    private final ConfigInstanceMapper configInstanceMapper;

    private final OperationLogService operationLogService;

    private static final int DEFAULT_QUEUE_CAPACITY = 1024;


    @Override
    public ConfigAllInfo findConfigAllInfo(String tpId, String itemId, String tenantId) {
        LambdaQueryWrapper<ConfigAllInfo> wrapper = Wrappers.lambdaQuery(ConfigAllInfo.class)
                .eq(StrUtil.isNotBlank(tpId), ConfigAllInfo::getTpId, tpId)
                .eq(StrUtil.isNotBlank(itemId), ConfigAllInfo::getItemId, itemId)
                .eq(StrUtil.isNotBlank(tenantId), ConfigAllInfo::getTenantId, tenantId);
        return configInfoMapper.selectOne(wrapper);
    }

    @Override
    public ConfigAllInfo findConfigRecentInfo(String... params) {
        ConfigAllInfo resultConfig;
        ConfigAllInfo configInstance = null;
        String instanceId = params[INDEX_3];
        if (StrUtil.isNotBlank(instanceId)) {
            LambdaQueryWrapper<ConfigInstanceInfo> instanceQueryWrapper = Wrappers.lambdaQuery(ConfigInstanceInfo.class)
                    .eq(ConfigInstanceInfo::getTpId, params[INDEX_0])
                    .eq(ConfigInstanceInfo::getItemId, params[INDEX_1])
                    .eq(ConfigInstanceInfo::getTenantId, params[INDEX_2])
                    .eq(ConfigInstanceInfo::getInstanceId, params[INDEX_3])
                    .orderByDesc(ConfigInstanceInfo::getGmtCreate)
                    .last("LIMIT 1");
            ConfigInstanceInfo instanceInfo = configInstanceMapper.selectOne(instanceQueryWrapper);
            if (instanceInfo != null) {
                String content = instanceInfo.getContent();
                configInstance = JSONUtil.toBean(content, ConfigAllInfo.class);
                configInstance.setContent(content);
                configInstance.setGmtCreate(instanceInfo.getGmtCreate());
                configInstance.setMd5(Md5Util.getTpContentMd5(configInstance));
            }
        }
        ConfigAllInfo configAllInfo = findConfigAllInfo(params[0], params[1], params[2]);
        if (configAllInfo == null && configInstance == null) {
            throw new ServiceException("Thread pool configuration is not defined");
        } else if (configAllInfo != null && configInstance == null) {
            resultConfig = configAllInfo;
        } else if (configAllInfo == null) {
            resultConfig = configInstance;
        } else {
            if (configAllInfo.getGmtModified().before(configInstance.getGmtCreate())) {
                resultConfig = configInstance;
            } else {
                resultConfig = configAllInfo;
            }
        }
        return resultConfig;
    }

    @Override
    public void insertOrUpdate(String identify, boolean isChangeNotice, ConfigAllInfo configAllInfo) {
        verification(identify);
        final LambdaQueryWrapper<ConfigAllInfo> queryWrapper = Wrappers.lambdaQuery(ConfigAllInfo.class)
                .eq(ConfigInfoBase::getTenantId, configAllInfo.getTenantId())
                .eq(ConfigInfoBase::getItemId, configAllInfo.getItemId())
                .eq(ConfigInfoBase::getTpId, configAllInfo.getTpId());
        final ConfigAllInfo existConfig = configInfoMapper.selectOne(queryWrapper);
        final ConfigServiceImpl configService = ApplicationContextHolder.getBean(this.getClass());
        configAllInfo.setCapacity(getQueueCapacityByType(configAllInfo));
        ConditionUtil.condition(
                existConfig == null,
                () -> configService.addConfigInfo(configAllInfo),
                () -> configService.updateConfigInfo(identify, isChangeNotice, configAllInfo)
        );
        if (isChangeNotice) {
            ConfigChangePublisher.notifyConfigChange(new LocalDataChangeEvent(identify, ContentUtil.getGroupKey(configAllInfo)));
        }
    }

    public void updateConfigInfo(String identify, boolean isChangeNotice, ConfigAllInfo config) {
        LambdaUpdateWrapper<ConfigAllInfo> wrapper = Wrappers.lambdaUpdate(ConfigAllInfo.class)
                .eq(ConfigAllInfo::getTpId, config.getTpId())
                .eq(ConfigAllInfo::getItemId, config.getItemId())
                .eq(ConfigAllInfo::getTenantId, config.getTenantId());
        config.setGmtCreate(null);
        config.setContent(ContentUtil.getPoolContent(config));
        config.setMd5(Md5Util.getTpContentMd5(config));
        recordOperationLog(config);
        try {
            // Create a temporary configuration of a thread pool configuration instance,
            // which can also be used as a historical configuration, but it is aimed at a single node.
            if (StrUtil.isNotBlank(identify)) {
                ConfigInstanceInfo instanceInfo = BeanUtil.convert(config, ConfigInstanceInfo.class);
                instanceInfo.setInstanceId(identify);
                configInstanceMapper.insert(instanceInfo);
            } else if (StrUtil.isEmpty(identify) && isChangeNotice) {
                List<String> identifyList = ConfigCacheService.getIdentifyList(config.getTenantId(), config.getItemId(), config.getTpId());
                if (CollectionUtil.isNotEmpty(identifyList)) {
                    for (String each : identifyList) {
                        ConfigInstanceInfo instanceInfo = BeanUtil.convert(config, ConfigInstanceInfo.class);
                        instanceInfo.setInstanceId(each);
                        configInstanceMapper.insert(instanceInfo);
                    }
                }
            } else {
                configInfoMapper.update(config, wrapper);
            }
        } catch (Exception ex) {
            log.error("[db-error] message: {}", ex.getMessage(), ex);
            throw ex;
        }
    }

    public Long addConfigInfo(ConfigAllInfo config) {
        config.setContent(ContentUtil.getPoolContent(config));
        config.setMd5(Md5Util.getTpContentMd5(config));
        Long configId = null;
        try {
            // Currently it is a single application, and it supports switching distributed locks during cluster deployment in the future.
            synchronized (ConfigService.class) {
                ConfigAllInfo configAllInfo = configInfoMapper.selectOne(
                        Wrappers.lambdaQuery(ConfigAllInfo.class)
                                .eq(ConfigAllInfo::getTpId, config.getTpId())
                                .eq(ConfigAllInfo::getDelFlag, DelEnum.NORMAL.getIntCode()));
                Assert.isNull(configAllInfo, "线程池配置已存在");
                if (SqlHelper.retBool(configInfoMapper.insert(config))) {
                    configId = config.getId();
                }
            }
        } catch (Exception ex) {
            log.error("[db-error] message: {}", ex.getMessage(), ex);
            throw ex;
        }
        return configId;
    }

    private void recordOperationLog(ConfigAllInfo requestParam) {
        LogRecordInfo logRecordInfo = LogRecordInfo.builder()
                .bizKey(requestParam.getItemId() + "_" + requestParam.getTpId())
                .bizNo(requestParam.getItemId() + "_" + requestParam.getTpId())
                .operator(Optional.of(UserContext.getUserName()).orElse("-"))
                .action(String.format("核心线程: %d, 最大线程: %d, 队列类型: %d, 队列容量: %d, 拒绝策略: %d", requestParam.getCoreSize(), requestParam.getMaxSize(), requestParam.getQueueType(),
                        requestParam.getCapacity(), requestParam.getRejectedType()))
                .category("THREAD_POOL_UPDATE")
                .detail(JSONUtil.toJsonStr(requestParam))
                .createTime(new Date())
                .build();
        operationLogService.record(logRecordInfo);
    }

    /**
     * <p> 不支持设置队列大小 {@link SynchronousQueue} {@link LinkedTransferQueue}
     *
     * @param config
     * @return
     */
    private Integer getQueueCapacityByType(ConfigAllInfo config) {
        int queueCapacity;
        if (LINKED_TRANSFER_QUEUE.getType().equals(config.getQueueType())) {
            queueCapacity = Integer.MAX_VALUE;
        } else {
            queueCapacity = config.getCapacity();
        }
        final List<Integer> queueTypes = Stream.of(
                ARRAY_BLOCKING_QUEUE.getType(),
                LINKED_BLOCKING_QUEUE.getType(),
                LINKED_BLOCKING_DEQUE.getType(),
                PRIORITY_BLOCKING_QUEUE.getType(),
                RESIZABLE_CAPACITY_LINKED_BLOCKING_QUEUE.getType()
        ).collect(Collectors.toList());
        boolean setDefaultFlag = queueTypes.contains(config.getQueueType()) && (config.getCapacity() == null || Objects.equals(config.getCapacity(), 0));
        if (setDefaultFlag) {
            queueCapacity = DEFAULT_QUEUE_CAPACITY;
        }
        return queueCapacity;
    }

    private void verification(String identify) {
        if (StrUtil.isNotBlank(identify)) {
            final Map<String, CacheItem> content = getContent(identify);
            Assert.isTrue(CollectionUtil.isNotEmpty(content), "If the thread pool instance does not exist, try page refresh");
        }
    }
}
